package com.wobito.system.mqtt;


import com.alibaba.fastjson.JSON;
import com.wobito.common.constants.BusinessException;
import com.wobito.common.constants.ErrorCode;
import com.wobito.common.utils.StringUtils;
import com.wobito.pojo.mqtt.MQTTBean;
import com.wobito.pojo.result.MQTTResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Slf4j
@Component
public class MQTTServer {

    @Resource
    private MQTTClientUtils mqttClientUtils;

    public CompletableFuture<String> publishAndWait(MQTTBean MQTTBean) {
        CompletableFuture<String> future = new CompletableFuture<>();
        // 发布消息
        mqttClientUtils.publish(MQTTBean.getTopic(), MQTTBean.getMsg());
        // 订阅并设置回调
        mqttClientUtils.subscribe(MQTTBean.getTopic() + "/Ack", 2, (topic, msg) -> {
            try {
                // 处理接收到的消息
                String response = new String(msg.getPayload(), StandardCharsets.UTF_8);
                System.out.println("收到消息："+topic+", 消息内容是："+ response);
                // 完成后设置CompletableFuture的结果
                future.complete(response);
            } catch (Exception e) {
                // 如果处理过程中发生异常，则设置CompletableFuture的异常
                future.completeExceptionally(e);
            }
        });
        return future;
    }

    public void publish(MQTTBean MQTTBean) {
        CompletableFuture<String> future = publishAndWait(MQTTBean);
        String r = null;
        try {
            r = future.get(); // 这将阻塞直到消息被接收或发生异常
            log.info("接收消息:{}",r);
            // 处理响应
            if (StringUtils.isEmpty(r)){
                throw new BusinessException(ErrorCode.FORBIDDEN,"请上传正确的人脸信息");
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
        MQTTResult mqttResult = JSON.parseObject(r, MQTTResult.class);
        if(!mqttResult.getInfo().getResult().equals("ok")){
            throw new BusinessException(ErrorCode.SYSTEM_ERROR,mqttResult.getInfo().getDetail());
        }
    }
}
